/*
 * Copyright (C) 2008 CERNET Network Center 
 * dvping/dvmcast/hdvping/hdvmcast series 
 * 
 * Design and coding: 
 *  	Xing Li <xing@cernet.edu.cn> 
 *	Congxiao Bao <congxiao@cernet.edu.cn>
 * 	Jinpeng Jiang <jjp02@mails.tsinghua.edu.cn>
 * 
 * Contributions:
 * 
 * Permission to use, copy, modify, and distribute this software for any
 * purpose with or without fee is hereby granted, provided that the above
 * copyright notice and this permission notice appear in all copies.
 */


#include "NetworkDVSource.h"
#include <math.h>

NetworkDVSource::NetworkDVSource(sa_family_t sa_family, in_port_t rcv_port, const vector<IPAddressFilter>& src_filters, int32_t log_period /*=0*/)
{
	this->sa_family = sa_family;
	this->rcv_port = rcv_port;
	this->src_filters = src_filters;
	this->log_period = log_period * 1000000LL;

	init_log_threshold = 0.5 / 8.0;		// 0.5Mbps
	fluctuate_threshold = 0.5 / 8.0;	// 0.5Mbps
}

bool NetworkDVSource::Init()
{
	if(group.IsValid())
	{
		sock = Socket::CreateUdpReceiver(group, rcv_port, true);
		if(source.IsValid())
		{
			if(!sock.JoinSourceGroup(source, group))
			{
				cout << "SSM join to (" << source << ", " << group << ") failed." << endl;
				return false;
			}
		}
		else
		{
			if(!sock.JoinGroup(group))
			{
				cout << "ASM join to " << group << " failed." << endl;
				return false;
			}
		}
	}
	else
	{
		sock = Socket::CreateUdpReceiver(sa_family, rcv_port, false);
	}
	sock.SetRecvBufferSize(1000000);
	log_start_time = DateTime::Now().GetTickCount();
	candi_sources.clear();
	tracked_sources.clear();
	return sock.IsValid();
}
DVPacket NetworkDVSource::GetPacket()
{
	WriteLog();
	DVPacket pkt;
	int32_t length = sock.ReceiveFrom(buffer, 65536, pkt.srcep, 100000);
	if(length < 4)
		return pkt;
	if(!src_filters.empty())
	{
		u_int32_t i = 0;
		for(i = 0; i < src_filters.size(); ++i)
		{
			if(src_filters[i].IsMatch(pkt.srcep.Address()))
				break;
		}
		if(i == src_filters.size())
			return pkt;			// emtpy packet
	}
	pkt.length = length;
	pktQueue.GetFreePacket(pkt);
	memcpy(pkt.pdata, buffer, pkt.length);
	OnPacketReceive(pkt);
	return pkt;
}
bool NetworkDVSource::Cleanup()
{
	return sock.Close();
}

bool NetworkDVSource::JoinGroup(const IPAddress& group)
{
	this->group = group;
	return true;
}
bool NetworkDVSource::JoinSourceGroup(const IPAddress& source, const IPAddress& group)
{
	this->source = source;
	this->group = group;
	return true;
}



bool NetworkDVSource::SetInitLogThreshold(double ilt)
{
	if(ilt < 0.0)
		return false;
	init_log_threshold = ilt;
	return true;
}
double NetworkDVSource::GetInitLogThreshold() const
{
	return init_log_threshold;
}
bool NetworkDVSource::SetFluctuateThreshold(double ft)
{
	if(ft <= 0.0)
		return false;
	fluctuate_threshold = ft;
	return true;
}
double NetworkDVSource::GetFluctuateThreshold() const
{
	return fluctuate_threshold;
}
void NetworkDVSource::OnPacketReceive(const DVPacket& pkt)
{
	if(log_period <= 0)
		return;

	int64_t now = DateTime::Now().GetTickCount();
	u_int32_t seq = ntohs(*(u_int16_t*)(buffer + 2));
	if(candi_sources.find(pkt.srcep) == candi_sources.end())
	{
		candi_sources[pkt.srcep].ep = pkt.srcep;
	}
	SourceStat& ss = candi_sources[pkt.srcep];
	if(ss.total_packets > 0)
	{
		if(seq < 1000 && ss.last_seq > 60000)
		{
			ss.lost_packets += (int)seq + 65536 - ss.last_seq - 1;
		}
		else if(seq > ss.last_seq)
		{
			ss.lost_packets += seq - ss.last_seq - 1;
		}
	}
	else
	{
		ss.start_time = now;
	}
	ss.bytes_recved += pkt.length;
	ss.total_packets ++;
	ss.last_seq = seq;
	ss.stop_time = now;
	return;
}
void NetworkDVSource::WriteLog()
{
	if(log_period <= 0)
		return;
	int64_t now = DateTime::Now().GetTickCount();
	if(now - log_start_time < log_period)
		return;
	// filter those candi_sources whose bandwidth less than min_threshold
	map<IPEndPoint, SourceStat>::iterator itc = candi_sources.begin();
	while(itc != candi_sources.end())
	{
		SourceStat& ss = itc->second;
		if(ss.stop_time > ss.start_time)
			ss.bandwidth = ss.bytes_recved / (double)(ss.stop_time - ss.start_time);
		else
			ss.bandwidth = ss.bytes_recved / (double)(now - ss.start_time);
		if(ss.bandwidth < init_log_threshold)
		{
			map<IPEndPoint, SourceStat>::iterator ittemp = itc;
			++itc;
			candi_sources.erase(ittemp);
		}
		else
		{
			++itc;
		}
	}
	// update tracked sources
	itc = candi_sources.begin();
	for(; itc != candi_sources.end(); ++itc)
	{
		SourceStat& ss = itc->second;
		if(tracked_sources.find(ss.ep) != tracked_sources.end())
		{
			NetworkSourceLog& nsl = tracked_sources[ss.ep];
			nsl.total_packets += ss.total_packets;
			nsl.lost_packets += ss.lost_packets;
			nsl.bytes_recved += ss.bytes_recved;
			nsl.stop_time = ss.stop_time;
			assert(!nsl.bandwidth.empty());
			if(fabs(nsl.bandwidth.back().second - ss.bandwidth) > fluctuate_threshold)
			{
				nsl.bandwidth.push_back(pair<int64_t, double>((ss.start_time + ss.stop_time) / 2, ss.bandwidth));
			}
		}
		else
		{
			NetworkSourceLog& nsl = tracked_sources[ss.ep];
			nsl.ep = ss.ep;
			nsl.start_time = ss.start_time;
			nsl.stop_time = ss.stop_time;
			nsl.lost_packets = ss.lost_packets;
			nsl.total_packets = ss.total_packets;
			nsl.bytes_recved = ss.bytes_recved;
			nsl.bandwidth.push_back(pair<int64_t, double>((ss.start_time + ss.stop_time) / 2, ss.bandwidth));
		}
	}
	// remove stopped sources
	map<IPEndPoint, NetworkSourceLog>::iterator itt = tracked_sources.begin();
	while(itt != tracked_sources.end())
	{
		if(candi_sources.find(itt->first) == candi_sources.end())
		{
            NetworkSourceLog& nsl = itt->second;
			dvlog.lock();
			dvlog << nsl << endl;
			dvlog.unlock();
			map<IPEndPoint, NetworkSourceLog>::iterator ittemp = itt;
			++itt;
			tracked_sources.erase(ittemp);
		}
		else
		{
			++itt;
		}
	}
	// restore candi_sources
	for(itc = candi_sources.begin(); itc != candi_sources.end(); ++itc)
	{
		SourceStat& ss = itc->second;
		ss.bandwidth = 0.0;
		ss.bytes_recved = 0;
		ss.lost_packets = 0;
		ss.start_time = 0;
		ss.stop_time = 0;
		ss.total_packets = 0;
		ss.last_seq = 0;
	}
	log_start_time = DateTime::Now().GetTickCount();
	return;
}

